---
title: Kotlin 工作流概述 | Kastrax 文档
description: Kastrax 中 Kotlin 工作流系统的详细介绍，包括工作流定义、步骤创建、变量处理和执行。
---

# Kotlin 工作流概述 ✅

Kastrax 的 Kotlin 工作流系统提供了一种强大的方式来组织和执行复杂的 AI 任务。本指南详细介绍了如何在 Kotlin 中使用工作流系统来创建和管理多步骤的 AI 流程。

## 工作流系统架构 ✅

Kastrax 工作流系统由以下核心组件组成：

1. **工作流接口 (Workflow)**：定义工作流的基本操作
2. **工作流步骤接口 (WorkflowStep)**：定义步骤的基本操作
3. **工作流上下文 (WorkflowContext)**：管理工作流执行状态
4. **变量引用 (VariableReference)**：引用工作流中的数据
5. **工作流执行引擎**：负责按照依赖关系执行步骤
6. **工作流构建器**：提供 DSL 用于创建工作流

## 基本用法 ✅

以下是在 Kastrax 中创建和使用工作流的基本示例：

```kotlin
import ai.kastrax.core.agent.agent
import ai.kastrax.core.workflow.workflow
import ai.kastrax.core.workflow.variable
import ai.kastrax.integrations.deepseek.deepSeek
import ai.kastrax.integrations.deepseek.DeepSeekModel
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
    // 创建代理
    val researchAgent = agent {
        name = "研究员"
        instructions = "你是一个研究员，负责收集和分析信息。"
        model = deepSeek {
            apiKey("your-deepseek-api-key")
            model(DeepSeekModel.DEEPSEEK_CHAT)
            temperature(0.7)
        }
    }
    
    val writingAgent = agent {
        name = "作家"
        instructions = "你是一个作家，负责创作高质量的内容。"
        model = deepSeek {
            apiKey("your-deepseek-api-key")
            model(DeepSeekModel.DEEPSEEK_CHAT)
            temperature(0.7)
        }
    }
    
    // 创建工作流
    val researchWorkflow = workflow("research-workflow", "研究并撰写文章") {
        // 研究步骤
        step(researchAgent) {
            id = "research"
            name = "研究"
            description = "研究主题"
            variables = mutableMapOf(
                "topic" to variable("$.input.topic")
            )
        }
        
        // 写作步骤
        step(writingAgent) {
            id = "writing"
            name = "写作"
            description = "基于研究撰写文章"
            after("research")
            variables = mutableMapOf(
                "research" to variable("$.steps.research.output.text")
            )
        }
        
        // 定义输出映射
        output {
            "researchResults" from "$.steps.research.output.text"
            "article" from "$.steps.writing.output.text"
            "wordCount" from {
                "$.steps.writing.output.text" to { text ->
                    (text as? String)?.split(" ")?.size ?: 0
                }
            }
        }
    }
    
    // 执行工作流
    val result = researchWorkflow.execute(mapOf(
        "topic" to "人工智能的未来"
    ))
    
    // 处理结果
    println("研究结果: ${result.output["researchResults"]}")
    println("文章: ${result.output["article"]}")
    println("字数: ${result.output["wordCount"]}")
}
```

## 工作流定义 ✅

使用 Kastrax 的 DSL 创建工作流：

```kotlin
val myWorkflow = workflow("my-workflow", "我的工作流描述") {
    // 步骤定义
    step(agent1) {
        id = "step1"
        name = "步骤 1"
        description = "第一个步骤"
        variables = mapOf(
            "input1" to variable("$.input.key1")
        )
    }
    
    step(agent2) {
        id = "step2"
        name = "步骤 2"
        description = "第二个步骤"
        after("step1")
        variables = mapOf(
            "input2" to variable("$.steps.step1.output.result")
        )
    }
    
    // 输出映射
    output {
        "result1" from "$.steps.step1.output.result"
        "result2" from "$.steps.step2.output.result"
        "combined" from {
            "$.steps.step2.output.result" to { result ->
                "处理结果: $result"
            }
        }
    }
}
```

## 步骤定义 ✅

工作流步骤是工作流的基本构建块，每个步骤可以执行特定的任务：

### 代理步骤

使用代理执行步骤：

```kotlin
step(myAgent) {
    id = "agent-step"
    name = "代理步骤"
    description = "使用代理执行任务"
    variables = mapOf(
        "input" to variable("$.input.query")
    )
}
```

### 函数步骤

使用函数执行步骤：

```kotlin
functionStep {
    id = "function-step"
    name = "函数步骤"
    description = "使用函数执行任务"
    variables = mapOf(
        "input" to variable("$.input.data")
    )
    execute { context ->
        val input = context.variables["input"] as? String ?: ""
        val result = processData(input)
        mapOf("result" to result)
    }
}
```

### 工具步骤

使用工具执行步骤：

```kotlin
toolStep(calculatorTool) {
    id = "tool-step"
    name = "工具步骤"
    description = "使用工具执行任务"
    variables = mapOf(
        "a" to variable("$.input.a"),
        "b" to variable("$.input.b")
    )
}
```

### 子工作流步骤

使用子工作流执行步骤：

```kotlin
subworkflowStep(subWorkflow) {
    id = "subworkflow-step"
    name = "子工作流步骤"
    description = "执行子工作流"
    variables = mapOf(
        "topic" to variable("$.input.topic")
    )
}
```

## 变量处理 ✅

Kastrax 工作流系统使用 JSONPath 风格的变量引用来访问工作流中的数据：

### 变量引用

```kotlin
// 引用输入数据
val inputRef = variable("$.input.key")

// 引用步骤输出
val stepOutputRef = variable("$.steps.step1.output.result")

// 引用全局变量
val variableRef = variable("$.variables.counter")

// 带默认值的引用
val refWithDefault = variable("$.input.optional", defaultValue = "默认值")
```

### 变量解析

工作流引擎在执行时解析变量引用：

1. 识别变量引用的源（输入、步骤、变量）
2. 使用路径导航到指定的属性
3. 应用可选的转换函数
4. 如果解析失败，使用默认值

## 控制流 ✅

Kastrax 工作流系统支持多种控制流模式：

### 顺序执行

使用 `after` 属性指定步骤的执行顺序：

```kotlin
step(agent1) {
    id = "step1"
    // ...
}

step(agent2) {
    id = "step2"
    after("step1")
    // ...
}

step(agent3) {
    id = "step3"
    after("step2")
    // ...
}
```

### 并行执行

不指定 `after` 属性的步骤可以并行执行：

```kotlin
step(agent1) {
    id = "step1"
    // ...
}

step(agent2) {
    id = "step2"
    // 不依赖于 step1，可以并行执行
    // ...
}

step(agent3) {
    id = "step3"
    after("step1", "step2")
    // 依赖于 step1 和 step2，等待它们完成后执行
    // ...
}
```

### 条件执行

使用条件步骤实现条件执行：

```kotlin
conditionalStep {
    id = "conditional-step"
    condition { context ->
        val value = context.variables["value"] as? Int ?: 0
        value > 10
    }
    thenStep {
        // 条件为真时执行的步骤
        step(agent1) {
            id = "then-step"
            // ...
        }
    }
    elseStep {
        // 条件为假时执行的步骤
        step(agent2) {
            id = "else-step"
            // ...
        }
    }
}
```

### 循环执行

使用循环步骤实现循环执行：

```kotlin
loopStep {
    id = "loop-step"
    maxIterations = 5
    condition { context, iteration ->
        val value = context.variables["value"] as? Int ?: 0
        value < 10 && iteration < maxIterations
    }
    body {
        step(agent1) {
            id = "loop-body"
            // ...
        }
    }
    afterLoop {
        step(agent2) {
            id = "after-loop"
            // ...
        }
    }
}
```

## 错误处理 ✅

Kastrax 工作流系统提供了多种错误处理机制：

### 步骤级错误处理

在步骤定义中添加错误处理：

```kotlin
step(agent1) {
    id = "risky-step"
    // ...
    onError {
        // 错误处理逻辑
        logger.error("步骤执行失败: ${it.message}")
        mapOf("error" to (it.message ?: "未知错误"))
    }
}
```

### 工作流级错误处理

在工作流定义中添加全局错误处理：

```kotlin
workflow("error-handling-workflow", "带错误处理的工作流") {
    // 步骤定义
    // ...
    
    // 全局错误处理
    onError { error, context ->
        logger.error("工作流执行失败: ${error.message}")
        
        // 返回错误结果
        mapOf(
            "success" to false,
            "error" to (error.message ?: "未知错误"),
            "partialResults" to context.steps.mapValues { it.value.output }
        )
    }
}
```

### 重试机制

为步骤添加重试策略：

```kotlin
step(agent1) {
    id = "retry-step"
    // ...
    retryStrategy {
        maxAttempts = 3
        initialDelay = 1000 // 毫秒
        maxDelay = 5000 // 毫秒
        multiplier = 2.0 // 指数退避
        retryOn = listOf(
            IllegalStateException::class,
            TimeoutException::class
        )
    }
}
```

## 高级功能 ✅

### 输出映射

使用输出映射转换工作流结果：

```kotlin
workflow("output-mapping-workflow", "带输出映射的工作流") {
    // 步骤定义
    // ...
    
    // 输出映射
    output {
        // 简单映射
        "result" from "$.steps.final-step.output.result"
        
        // 带转换的映射
        "formattedResult" from {
            "$.steps.final-step.output.result" to { result ->
                "结果: $result"
            }
        }
        
        // 组合多个值
        "summary" from {
            listOf(
                "$.steps.step1.output.result",
                "$.steps.step2.output.result"
            ) to { values ->
                "步骤 1: ${values[0]}, 步骤 2: ${values[1]}"
            }
        }
    }
}
```

### 工作流组合

组合多个工作流创建复杂流程：

```kotlin
// 创建子工作流
val subWorkflow1 = workflow("sub-workflow-1", "子工作流 1") {
    // ...
}

val subWorkflow2 = workflow("sub-workflow-2", "子工作流 2") {
    // ...
}

// 创建主工作流
val mainWorkflow = workflow("main-workflow", "主工作流") {
    // 使用子工作流
    subworkflowStep(subWorkflow1) {
        id = "sub1"
        variables = mapOf(
            "input" to variable("$.input.data")
        )
    }
    
    subworkflowStep(subWorkflow2) {
        id = "sub2"
        after("sub1")
        variables = mapOf(
            "input" to variable("$.steps.sub1.output.result")
        )
    }
    
    // 输出映射
    output {
        "finalResult" from "$.steps.sub2.output.result"
    }
}
```

### 工作流模板

创建可重用的工作流模板：

```kotlin
// 创建工作流模板
val workflowTemplate = workflowTemplate("template-name", "模板描述") {
    // 定义参数
    parameter("agentType", "代理类型", String::class)
    parameter("iterations", "迭代次数", Int::class, defaultValue = 3)
    
    // 模板构建函数
    build { params ->
        val agentType = params["agentType"] as String
        val iterations = params["iterations"] as Int
        
        // 根据参数创建工作流
        workflow("generated-workflow", "从模板生成的工作流") {
            // 使用参数配置工作流
            // ...
            
            // 根据迭代次数创建步骤
            for (i in 1..iterations) {
                step(getAgentByType(agentType)) {
                    id = "step-$i"
                    // ...
                }
            }
        }
    }
}

// 使用模板创建工作流
val myWorkflow = workflowTemplate.createWorkflow(mapOf(
    "agentType" to "researcher",
    "iterations" to 5
))
```

### 工作流监控

监控工作流执行状态：

```kotlin
// 创建工作流引擎
val workflowEngine = WorkflowEngine()

// 注册工作流
workflowEngine.registerWorkflow("my-workflow", myWorkflow)

// 监听工作流状态更新
workflowEngine.stateUpdates.collect { state ->
    println("工作流 ${state.workflowId} 状态: ${state.status}")
    println("步骤状态:")
    state.steps.forEach { (stepId, stepState) ->
        println("  $stepId: ${stepState.status}")
    }
}

// 执行工作流
val result = workflowEngine.executeWorkflow(
    workflowId = "my-workflow",
    input = mapOf("topic" to "人工智能")
)
```

## 与其他组件集成 ✅

### 与代理集成

工作流可以与 Kastrax 代理无缝集成：

```kotlin
// 创建代理
val myAgent = agent {
    name = "助手"
    instructions = "你是一个有帮助的助手。"
    model = deepSeek {
        apiKey("your-deepseek-api-key")
        model(DeepSeekModel.DEEPSEEK_CHAT)
    }
}

// 在工作流中使用代理
val workflow = workflow("agent-workflow", "使用代理的工作流") {
    step(myAgent) {
        id = "agent-step"
        variables = mapOf(
            "query" to variable("$.input.query")
        )
    }
    
    output {
        "response" from "$.steps.agent-step.output.text"
    }
}
```

### 与工具集成

工作流可以与 Kastrax 工具集成：

```kotlin
// 创建工具
val calculatorTool = tool("calculator") {
    description("执行数学计算")
    parameters {
        parameter("expression", "数学表达式", String::class)
    }
    execute { params ->
        val expression = params["expression"] as String
        val result = evaluateExpression(expression)
        "结果: $result"
    }
}

// 在工作流中使用工具
val workflow = workflow("tool-workflow", "使用工具的工作流") {
    toolStep(calculatorTool) {
        id = "calculator-step"
        variables = mapOf(
            "expression" to variable("$.input.expression")
        )
    }
    
    output {
        "result" from "$.steps.calculator-step.output"
    }
}
```

### 与 RAG 集成

工作流可以与 Kastrax RAG 系统集成：

```kotlin
// 创建 RAG 系统
val embeddingService = FastEmbeddingService.create()
val vectorStore = InMemoryVectorStore()
val rag = RAG(vectorStore, embeddingService)

// 加载文档
val loader = DocumentLoader.fromFile("knowledge_base.pdf")
val documents = loader.load()
rag.addDocuments(documents)

// 在工作流中使用 RAG
val workflow = workflow("rag-workflow", "使用 RAG 的工作流") {
    functionStep {
        id = "search-step"
        variables = mapOf(
            "query" to variable("$.input.query")
        )
        execute { context ->
            val query = context.variables["query"] as? String ?: ""
            val results = rag.search(query, limit = 3)
            val context = results.joinToString("\n\n") { it.document.content }
            mapOf("context" to context)
        }
    }
    
    step(myAgent) {
        id = "answer-step"
        after("search-step")
        variables = mapOf(
            "query" to variable("$.input.query"),
            "context" to variable("$.steps.search-step.output.context")
        )
    }
    
    output {
        "answer" from "$.steps.answer-step.output.text"
    }
}
```

## 总结 ✅

Kastrax 的 Kotlin 工作流系统提供了一个强大而灵活的框架，用于构建复杂的 AI 任务流程。通过组合代理、工具和其他组件，您可以创建高度自定义的 AI 应用程序，满足各种业务需求。

无论您是构建简单的顺序流程还是复杂的多步骤应用，Kastrax 工作流系统都能提供所需的工具和组件，帮助您创建更强大、更智能的 AI 解决方案。
